package com.github.ghsea.scheduler.core;

import com.github.ghsea.scheduler.util.NetUtils;
import com.github.ghsea.scheduler.zkclient.ChildrenChangedListener;
import com.github.ghsea.scheduler.zkclient.ZkClient;
import java.util.Collection;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.leader.LeaderLatch;
import org.apache.curator.framework.recipes.leader.Participant;
import org.apache.zookeeper.data.Stat;

/**
 * 多台机器启动时，选举Leader，并将leader机器的IP写入zookeeper的临时节点
 * 1.所有机器都监控zk上的leader临时节点，如果leader挂掉，重新选主
 * 2.leader发现有机器的临时节点掉线时，拿到掉线的IP,间隔3分钟，最多check3次，如果发现掉线的机器没有上线，即判定机器宕机。
 * 进入rebalance阶段，通过下线机器的IP去数据库查出task,并在本地启动timer。这会导致leader机器压力较大，特别是在机器较多时，压力不能分摊，由于暂时gis的机器较少，目前不考虑这个问题。
 * 3.机器启动时，拿着ip去mysql表查出task,并在本地启动timer。
 *
 * 本方案的缺陷
 * A:leader是机器A，机器B宕机,机器A对机器B进行状态检查并rebalance的过程中，在rebalance完成之前机器A也宕机了。这种情况下
 * 机器B的任务节点将不会被执行。
 * B.宕机的Task全部被leader接管，task分配会有一些倾斜
 */
//@Component
@Slf4j
class ZookeeperHaTaskScheduler {

    private String zkServers;

    private ZkClient zkClient;

    private CuratorFramework curatorFramework;

    private boolean isLeader = false;

    private final String PATH_ROOT = "/gis";

    private final String PATH_LEADER = PATH_ROOT + "/leader";

    private final String PATH_CLUSTER = PATH_ROOT + "/cluster";

    private LeaderLatch leaderLatch;

    /**
     * 当前机器的id,使用IP进行标识
     */
    private String serverId;

    private int isLeaderCount = 0;

    private int isSlaveCount = 0;

    private final int LEADER_CHECK_TIMES = 3;

    private ChildrenChangedListener serverListener;

    @Resource
    private TaskScheduler scheduler;

    @Resource(name = "scheduledExecutor")
    private ScheduledExecutorService scheduledExecutor;

    @Resource(name = "tinyTaskExecutor")
    private ExecutorService tinyTaskExecutor;

    @PostConstruct
    void setUp() throws Exception {
        zkClient = ZkClient.getInstance(zkServers);
        curatorFramework = zkClient.getClient();

        serverId = NetUtils.getLocalHost();
        leaderLatch = new LeaderLatch(curatorFramework, PATH_LEADER, serverId);
        leaderLatch.start();
        registerToCluster();
        selectLeader();
    }

    @PreDestroy
    void destroy() {
        zkClient.close();
    }

    private void selectLeader() {
        scheduledExecutor.scheduleAtFixedRate(() -> {
            try {
                //首先利用serverId检查自己是否还存在于leaderlatch选举结果集中，考虑网络阻塞，zk数据异常丢失等情况
                Collection<Participant> participants = leaderLatch.getParticipants();
                boolean isExist = participants.stream().anyMatch(p -> serverId.equals(p.getId()));
                if (!isExist) {
                    leaderLatch.close();
                    leaderLatch = new LeaderLatch(curatorFramework, PATH_LEADER, serverId);
                    leaderLatch.start();
                }

                //查看当前leader是否是自己
                Participant leader = leaderLatch.getLeader();
                boolean hasLeaderShip = serverId.equals(leader.getId());

                //主从切换缓冲,Leader临时挂掉了(比如因网络原因导致心跳失败)不能认为其挂掉了，需经过多次check,都执行主从切换
                if (hasLeaderShip) {
                    ++isLeaderCount;
                    isSlaveCount = 0;
                } else {
                    isLeaderCount = 0;
                    ++isSlaveCount;
                }

                if (isLeaderCount >= LEADER_CHECK_TIMES && !isLeader) {
                    log.info("Currently run as leader serverId= {}", serverId);
                    isLeader = true;
                    watchCluster();
                }

                if (isSlaveCount >= LEADER_CHECK_TIMES && isLeader) {
                    log.info("Currently run as slave serverId= {}", serverId);
                    isLeader = false;
                    cancelWatchCluster();
                }
            } catch (Exception e) {
                log.error(e.getMessage(), e);
            }
        }, 1, 1, TimeUnit.MINUTES);
    }

    private void registerToCluster() {
        String zkPath = PATH_CLUSTER + "/" + serverId;
        try {
            zkClient.createEphemeral(zkPath, null, true);
        } catch (Exception e) {
            log.error(e.getMessage(),e);
            throw new RuntimeException(e);
        }

        log.info("registered  server {}", zkPath);
    }

    /**
     * 监控/cluster子节点，当发现有节点掉线时，将它的task重新调度
     */
    private void watchCluster() throws Exception {
        if (!isLeader) {
            return;
        }

        serverListener = new ChildrenChangedListener() {
            @Override
            public void notifyChildAdded(String childAdded, Object data) {
            }

            @Override
            public void notifyChildRemoved(String childRemoved) {
                rescheduleIfServerDown(childRemoved);
            }

            @Override
            public void notifyChildUpdated(String childUpdated, Stat stat, Object data) {
            }
        };
        zkClient.getChildrenAndWatch(PATH_CLUSTER, serverListener);
    }

    private void rescheduleIfServerDown(String serverZkPath) {
        tinyTaskExecutor.execute(() -> {
            if (isServerDown(serverZkPath)) {
                String downSeverId = computChildPath(serverZkPath);
                scheduler.initTaskAndSchedule(downSeverId,true);
            }
        });
    }

    /**
     * 取消监控/cluster子节点
     */
    private void cancelWatchCluster() {
        if (serverListener == null) {
            return;
        }

        zkClient.removeListener(PATH_CLUSTER, serverListener);
    }

    /**
     * 检查从zookeeper上掉线的机器是否真的停机了。这里排除机器重启的情况。机器重启时，它会重新初始化之前属于它的task
     *
     * @param serverZkPath 服务器在zk上的完整路径,形如:/gis/cluster/192.168.0.1
     * @return 服务器是否已宕机
     */
    private boolean isServerDown(String serverZkPath) {
        final int maxCheckTimes = 3;
        int maxZkFailTimes = 3;
        int zkFailTimes = 0;
        for (int i = 0; i < maxCheckTimes; i++) {
            try {
                if (zkClient.exist(serverZkPath)) {
                    return false;
                }

                //考虑机器重启的情况，休眠时间不能太短。
                TimeUnit.MINUTES.sleep(2);
            } catch (Exception e) {
                zkFailTimes++;
                if (zkFailTimes <= maxZkFailTimes) {
                    i--;
                }

                log.error(e.getMessage(), e);
            }
        }

        return true;
    }

    /**
     * 从完整的ZK路径拿到最下一级的路径
     * /gis/servers/192.168.0.0   ->192.168.0.0
     *
     * @param fullZkPath 完整的zk路径
     * @return 最下层的子路径
     */
    private String computChildPath(String fullZkPath) {
        return StringUtils.substring(fullZkPath, StringUtils.lastIndexOf(fullZkPath, "/") + 1);
    }

}
